/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package com.vng.zing.zprofile.thrift.wrapper;

import com.vng.zing.calc.thrift.Profile;
import com.vng.zing.common.CompressMed;
import com.vng.zing.common.IZStructor;
import com.vng.zing.common.ZCommonDef;
import com.vng.zing.configer.ZConfig;
import com.vng.zing.jni.zicache.ZCacheBase.LoadWorker;
import com.vng.zing.jni.zicache.ZiCache;
import com.vng.zing.logger.ZLogger;
import com.vng.zing.storage.MapDataResult;
import com.vng.zing.storage.ValueResult;
import com.vng.zing.zcommon.thrift.ECode;
import java.util.List;
import java.util.Map;
import org.apache.log4j.Logger;

/**
 *
 * @author namnq
 */
public class ZProfileClientWithCache {

	private static final Logger _Logger = ZLogger.getLogger(ZProfileClientWithCache.class);
	protected final String _name;
	protected ZiCache<Integer/*KType*/, Profile/*VType*/> _innerCache;
	protected ZProfileClient _innerClient;
	protected final IZStructor<Integer/*KType*/> _keyStructor = new IZStructor.IntegerStructor/*KStructor*/();
	protected final IZStructor<Profile/*VType*/> _valueStructor = new TValueStructor();

	public static class TValueStructor implements IZStructor<Profile/*VType*/> {

		@Override
		public Profile/*VType*/ ctor() {
			return new Profile/*VType*/();
		}
	}

	public static class TValueLoadCb implements ZiCache.ILoadCallback<Integer/*KType*/, Profile/*VType*/> {

		private final ZProfileClient _client;

		public TValueLoadCb(ZProfileClient client) {
			_client = client;
			assert (_client != null);
		}

		@Override
		public ValueResult<Profile/*VType*/> perform(LoadWorker<Integer/*KType*/, Profile/*VType*/> sender, Integer/*KType*/ key) {
			Profile vr = _client.getProfile(key);
                        _Logger.info("CACHED from DB with id" + key);
                        if(vr==null){
                            return  ValueResult.ctor(null, -ECode.C_FAIL.getValue());
                        }else{
                            return ValueResult.ctor(vr, ECode.C_SUCCESS.getValue());
                        }
			
		}
	}

//	public static class TValueMultiLoadCb implements ZiCache.IMultiLoadCallback<Integer/*KType*/, Profile/*VType*/> {
//
//		private final ZProfileClient _client;
//
//		public TValueMultiLoadCb(ZProfileClient client) {
//			_client = client;
//			assert (_client != null);
//		}
//
//		@Override
//		public MapDataResult<Integer/*KType*/, Profile/*VType*/> perform(LoadWorker<Integer/*KType*/, Profile/*VType*/> sender, Map<Integer/*KType*/, Integer> _errorMap, List<Integer/*KType*/> keys) {
//			TMapDataResult mdr = _client.mmultiGet(keys);
//			if (_errorMap != null && mdr.errorMap != null && !mdr.errorMap.isEmpty()) {
//				_errorMap.putAll(mdr.errorMap);
//			}
//			return MapDataResult.ctor(mdr.isSetDataMap() ? mdr.dataMap : null, mdr.error);
//		}
//	}

	public ZProfileClientWithCache(String name) {
		_name = name;
		initClient();
		initCache();
	}

	public ZProfileClient getInnerClient() {
		return _innerClient;
	}

	private void initClient() {
		_innerClient = new ZProfileClient(_name);
	}

	private void initCache() {
		assert (_innerClient != null);
		uninitCache();
		long tblSize = ZConfig.Instance.getLong(ZProfileClientWithCache.class, _name, "tbl_size", ZCommonDef.CacheTblSizeDefault);
		long lruSize = ZConfig.Instance.getLong(ZProfileClientWithCache.class, _name, "lru_size", ZCommonDef.CacheLruSizeDefault);
		long maxNfSize = ZConfig.Instance.getLong(ZProfileClientWithCache.class, _name, "max_size_queue_load", ZCommonDef.CacheMaxSizeQueueLoadDefault);
		int nloadsAtime = ZConfig.Instance.getInt(ZProfileClientWithCache.class, _name, "nloads_atime", ZCommonDef.CacheNLoadsAtimeDefault);
		int nthreadsLoad = ZConfig.Instance.getInt(ZProfileClientWithCache.class, _name, "nthreads_load", ZCommonDef.CacheNThreadsLoadDefault);
		int expSeconds = ZConfig.Instance.getInt(ZProfileClientWithCache.class, _name, "item_exp_seconds", ZCommonDef.CacheItemExpSecondsDefault);
		String strCompressMed = ZConfig.Instance.getString(ZProfileClientWithCache.class, _name, "compress_med", ZCommonDef.CacheCompressMedDefault.toString());
		CompressMed compressMed = CompressMed.findByName(strCompressMed);
		if (compressMed == null) {
			compressMed = CompressMed.NOCOMP;
		}
		int thresholdComp = ZConfig.Instance.getInt(ZProfileClientWithCache.class, _name, "threshold_comp", ZCommonDef.CacheThresholdCompDefault);
		//
		_innerCache = new ZiCache<Integer/*KType*/, Profile/*VType*/>(_keyStructor, _valueStructor, tblSize, lruSize);
		_innerCache.setLoadCallback(new TValueLoadCb(_innerClient));
		//_innerCache.setMultiLoadCallback(new TValueMultiLoadCb(_innerClient));
		//
		_innerCache.setMaxNfSize(maxNfSize); //native invoke
		_innerCache.setNumLoadsAtime(nloadsAtime);
		_innerCache.setNumThreadsLoad(nthreadsLoad);
		_innerCache.setExpTime(expSeconds); //native invoke
		_innerCache.setCompressMed(compressMed); //native invoke
		_innerCache.setThresholdComp(thresholdComp); //native invoke
		//
		//
		tblSize = _innerCache.getTblSize();
		lruSize = _innerCache.getLruSize();
		maxNfSize = _innerCache.getMaxNfSize();
		nloadsAtime = _innerCache.getNumLoadsAtime();
		nthreadsLoad = _innerCache.getNumThreadsLoad();
		expSeconds = _innerCache.getExpTime();
		compressMed = _innerCache.getCompressMed();
		thresholdComp = _innerCache.getThresholdComp();
		//
		_Logger.info("initCache("
				+ "tblSize=" + tblSize
				+ ", lruSize=" + lruSize
				+ ", maxNfSize=" + maxNfSize
				+ ", nloadsAtime=" + nloadsAtime
				+ ", nthreadsLoad=" + nthreadsLoad
				+ ", expSeconds=" + expSeconds
				+ ", compressMed=" + compressMed
				+ ", thresholdComp=" + thresholdComp
				+ ")");
	}

	private void uninitCache() {
		if (_innerCache != null) {
			_innerCache.stopThreadsLoad();
			_innerCache = null;
		}
	}

	///~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
	///statistic & config methods
	///
	/**
	 * setMaxNfSize
	 *
	 * @param Số load-notification tối đa chứa trong queue. Khi số
	 * load-notification vượt qua số này, lệnh enqueue sẽ trả về mã lỗi overload
	 */
	////@Override
	public void setMaxNfSize(long maxNfSize) {
		_innerCache.setMaxNfSize(maxNfSize);
	}

	/**
	 * getMaxNfSize
	 *
	 * @return Số load-notification tối đa chứa trong queue. Khi số
	 * load-notification vượt qua số này, lệnh enqueue sẽ trả về mã lỗi overload
	 */
	////@Override
	public long getMaxNfSize() {
		return _innerCache.getMaxNfSize();
	}

	/**
	 * queueLoadSize
	 *
	 * @return Số load-notification hiện có trong queue
	 */
	////@Override
	public long queueLoadSize() {
		return _innerCache.queueLoadSize();
	}

	/**
	 * setNumLoadsAtime
	 *
	 * @param Số item được load cùng lúc tối đa trong mỗi lần thực thi (gọi
	 * multiLoadCallback, thường là trong multiLoadCallback sẽ gọi multiGet đến
	 * storage service)
	 */
	////@Override
	public void setNumLoadsAtime(int nloadsAtime) {
		_innerCache.setNumLoadsAtime(nloadsAtime);
	}

	/**
	 * getNumLoadsAtime
	 *
	 * @return Số item được load cùng lúc tối đa trong mỗi lần thực thi (gọi
	 * multiLoadCallback, thường là trong multiLoadCallback sẽ gọi multiGet đến
	 * storage service)
	 */
	////@Override
	public int getNumLoadsAtime() {
		return _innerCache.getNumLoadsAtime();
	}

	/**
	 * clearAll Xóa load queue & wakeup các load-thread đang chờ + Xóa tất cả
	 * items trong cache
	 */
	////@Override
	public void clearAll() {
		_innerCache.clearAll();
	}

	/**
	 * getStat
	 *
	 * @return Các thống kê của cache & queue
	 */
	////@Override
	public com.vng.zing.jni.zicache.ZCacheBase.Stat getStat() {
		return _innerCache.getStat();
	}

	/**
	 * clearStat Xóa Các thống kê của cache & queue
	 */
	////@Override
	public void clearStat() {
		_innerCache.clearStat();
	}

	/**
	 * getTblSize
	 *
	 * @return Kích thước bảng băm của cache. TblSize nên lớn hơn hoặc bằng
	 * LruSize để tránh tình trạng đụng độ item (collision - được chèn vào cùng
	 * vị trí trong bảng băm) ảnh hưởng đến tốc độ khi tìm kiếm item
	 */
	////@Override
	public long getTblSize() {
		return _innerCache.getTblSize();
	}

	/**
	 * getLruSize
	 *
	 * @return Số item tối đa chứa trong cache. Quá số lượng item này, các item
	 * ít được truy xuất nhất sẽ tự động bị xóa
	 */
	////@Override
	public long getLruSize() {
		return _innerCache.getLruSize();
	}

	/**
	 * setExpTime
	 *
	 * @param Thời gian tồn tại chung của item trong trong cache (đơn vị: giây),
	 * quá thời gian này item sẽ bị xóa 1 cách thụ động (khi truy xuất)
	 */
	////@Override
	public void setExpTime(int expSeconds) {
		_innerCache.setExpTime(expSeconds);
	}

	/**
	 * getExpTime
	 *
	 * @return Thời gian tồn tại chung của item trong trong cache (đơn vị:
	 * giây), quá thời gian này item sẽ bị xóa 1 cách thụ động (khi truy xuất)
	 */
	////@Override
	public int getExpTime() {
		return _innerCache.getExpTime();
	}

	/**
	 * setCompressMed
	 *
	 * @param Phương pháp nén item trong cache
	 */
	////@Override
	public void setCompressMed(CompressMed compressMed) {
		_innerCache.setCompressMed(compressMed);
	}

	/**
	 * getCompressMed
	 *
	 * @return Phương pháp nén item trong cache
	 */
	////@Override
	public CompressMed getCompressMed() {
		return _innerCache.getCompressMed();
	}

	/**
	 * setThresholdComp
	 *
	 * @param Ngưỡng nén item trong cache. Khi kích thước của item vượt qua
	 * ngưỡng này thì item sẽ được nén lại
	 */
	////@Override
	public void setThresholdComp(int thresholdComp) {
		_innerCache.setThresholdComp(thresholdComp);
	}

	/**
	 * getThresholdComp
	 *
	 * @return Ngưỡng nén item trong cache. Khi kích thước của item vượt qua
	 * ngưỡng này thì item sẽ được nén lại
	 */
	////@Override
	public int getThresholdComp() {
		return _innerCache.getThresholdComp();
	}

	/**
	 * size
	 *
	 * @return Số lượng item hiện tại có trong cache
	 */
	////@Override
	public long size() {
		return _innerCache.size();
	}

	///~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
	///cache-read data methods
	///
	/**
	 * getItemTimeExp
	 *
	 * @param key
	 * @return Thời gian (đơn vị: giây) còn lại đến lúc hết hạn của item trong
	 * cache, lớn hơn 0 nếu còn, bé hơn không nếu item đã hết hạn
	 */
	////@Override
	public int getItemTimeExp(Integer/*KType*/ key) {
		return _innerCache.getItemTimeExp(key);
	}

	/**
	 * extItemTimeExp
	 *
	 * @param key
	 * @param expSeconds
	 * @return Gia thời hạn (cộng thêm thời gian - đơn vị: giây) cho item trong
	 * cache
	 */
	////@Override
	public boolean extItemTimeExp(Integer/*KType*/ key, int expSeconds) {
		return _innerCache.extItemTimeExp(key, expSeconds);
	}

	/**
	 * getCache
	 *
	 * @param key
	 * @return Đọc item trong cache
	 */
	//////@Override
	public Profile/*VType*/ getCache(Integer/*KType*/ key) {
		return _innerCache.get(key);
	}

	/**
	 * existCache
	 *
	 * @param key
	 * @return Kiểm tra item có trong cache, item hết hạn hoặc đang được load
	 * (chưa load xong) được xem như không tồn tại
	 *
	 */
	//////@Override
	public boolean existCache(Integer/*KType*/ key) {
		return _innerCache.exist(key, true);
	}

	/**
	 * multiGetCache
	 *
	 * @param keys
	 * @return Đọc nhiều item trong cache
	 */
	//////@Override
	public Map<Integer/*KType*/, Profile/*VType*/> multiGetCache(List<Integer/*KType*/> keys) {
		return _innerCache.multiGet(keys);
	}

	/**
	 * multiGetCache
	 *
	 * @param _outFailKeys
	 * @param keys
	 * @return Đọc nhiều item trong cache
	 */
	//////@Override
	public Map<Integer/*KType*/, Profile/*VType*/> multiGetCache(List<Integer/*KType*/> _outFailKeys, List<Integer/*KType*/> keys) {
		return _innerCache.multiGet(_outFailKeys, keys);
	}

	/**
	 * multiExistCache
	 *
	 * @param keys
	 * @return Kiểm tra tồn tại nhiều item trong cache
	 */
	//////@Override
	public List<Integer/*KType*/> multiExistCache(List<Integer/*KType*/> keys) {
		return _innerCache.multiExist(keys);
	}

	/**
	 * multiExistCache
	 *
	 * @param _outFailKeys
	 * @param keys
	 * @return Kiểm tra tồn tại nhiều item trong cache
	 */
	//////@Override
	public List<Integer/*KType*/> multiExistCache(List<Integer/*KType*/> _outFailKeys, List<Integer/*KType*/> keys) {
		return _innerCache.multiExist(_outFailKeys, keys);
	}

	///~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
	///read-load data methods
	///
	/**
	 * getaLoad
	 *
	 * @param key
	 * @return get item from cache and then load it from remote storage if not
	 * exist in cache
	 */
	////@Override
	public Profile/*VType*/ getaLoad(Integer/*KType*/ key) {
                _Logger.info("READ from cache with id" + key);
		return _innerCache.getaLoad(key);
	}

	public ValueResult<Profile/*VType*/> getaLoadWithEC(Integer/*KType*/ key) {
		return _innerCache.getaLoadWithEC(key);
	}

	/**
	 * multiGetaLoad
	 *
	 * @param keys
	 * @return multi-get items from cache and then load them from remote storage
	 * if not exist in cache
	 */
	////@Override
	public Map<Integer/*KType*/, Profile/*VType*/> multiGetaLoad(List<Integer/*KType*/> keys) {
		return _innerCache.multiGetaLoad(keys);
	}

	/**
	 * multiGetaLoad
	 *
	 * @param _outFailKeys
	 * @param keys
	 * @return multi-get items from cache and then load them from remote storage
	 * if not exist in cache
	 */
	////@Override
	public Map<Integer/*KType*/, Profile/*VType*/> multiGetaLoad(List<Integer/*KType*/> _outFailKeys, List<Integer/*KType*/> keys) {
		return _innerCache.multiGetaLoad(_outFailKeys, keys);
	}

	public MapDataResult<Integer/*KType*/, Profile/*VType*/> multiGetaLoadWithEC(Map<Integer/*KType*/, Integer> _errorMap, List<Integer/*KType*/> keys) {
		return _innerCache.multiGetaLoadWithEC(_errorMap, keys);
	}

	/**
	 * getaLoadBg
	 *
	 * @param key
	 * @return get item from cache and then asynchronous-load it (background)
	 * from remote storage if not exist in cache
	 */
	////@Override
	public Profile/*VType*/ getaLoadBg(Integer/*KType*/ key) {
		return _innerCache.getaLoadBg(key);
	}

	/**
	 * multiGetaLoadBg
	 *
	 * @param keys
	 * @return multi-get items from cache and then asynchronous-load them
	 * (background) from remote storage if not exist in cache
	 */
	////@Override
	public Map<Integer/*KType*/, Profile/*VType*/> multiGetaLoadBg(List<Integer/*KType*/> keys) {
		return _innerCache.multiGetaLoadBg(keys);
	}

	/**
	 * multiGetaLoadBg
	 *
	 * @param _outFailKeys
	 * @param keys
	 * @return multi-get items from cache and then asynchronous-load them
	 * (background) from remote storage if not exist in cache
	 */
	////@Override
	public Map<Integer/*KType*/, Profile/*VType*/> multiGetaLoadBg(List<Integer/*KType*/> _outFailKeys, List<Integer/*KType*/> keys) {
		return _innerCache.multiGetaLoadBg(_outFailKeys, keys);
	}

	/**
	 * loadBg
	 *
	 * @param key
	 * @return asynchronous-load item (background) from remote storage then put
	 * back into cache
	 */
	////@Override
	public boolean loadBg(Integer/*KType*/ key) {
		return _innerCache.loadBg(key);
	}

	/**
	 * multiLoadBg
	 *
	 * @param keys
	 * @return asynchronous-load items (background) from remote storage then put
	 * back into cache
	 */
	////@Override
	public boolean multiLoadBg(List<Integer/*KType*/> keys) {
		return _innerCache.multiLoadBg(keys);
	}

	///~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
	///cache-write data methods
	///
	/**
	 * putCache
	 *
	 * @param key
	 * @param value
	 * @param policy
	 * @return put item into cache
	 */
	//////@Override
	public boolean putCache(Integer/*KType*/ key, Profile/*VType*/ value, com.vng.zing.zcommon.thrift.PutPolicy policy) {
                _Logger.info("PUT into cache with id" + key);
		return _innerCache.put(key, value, com.vng.zing.common.PutPolicy.fromTPutPolicy(policy));
	}

	/**
	 * putCache
	 *
	 * @param key
	 * @param value
	 * @param expSeconds
	 * @param policy
	 * @return put item into cache with specified number of second expired
	 */
	//////@Override
	public boolean putCache(Integer/*KType*/ key, Profile/*VType*/ value, int expSeconds, com.vng.zing.zcommon.thrift.PutPolicy policy) {
		return _innerCache.put(key, value, true, expSeconds, com.vng.zing.common.PutPolicy.fromTPutPolicy(policy));
	}

	/**
	 * popCache
	 *
	 * @param key
	 * @return pop item from cache (get then remove)
	 */
	//////@Override
	public Profile/*VType*/ popCache(Integer/*KType*/ key) {
		return _innerCache.pop(key);
	}

	/**
	 * removeCache
	 *
	 * @param key
	 * @return remove item from cache
	 */
	//////@Override
	public boolean removeCache(Integer/*KType*/ key) {
            _Logger.info("REMOVE into cache with id" + key);
		return _innerCache.remove(key);
	}

	/**
	 * multiPutCache
	 *
	 * @param dtMap
	 * @param policy
	 * @return multi-put items into cache
	 */
	//////@Override
	public boolean multiPutCache(Map<Integer/*KType*/, Profile/*VType*/> dtMap, com.vng.zing.zcommon.thrift.PutPolicy policy) {
		return _innerCache.multiPut(dtMap, com.vng.zing.common.PutPolicy.fromTPutPolicy(policy));
	}

	/**
	 * multiPutCache
	 *
	 * @param _outFailKeys
	 * @param dtMap
	 * @param policy
	 * @return multi-put items into cache
	 */
	//////@Override
	public boolean multiPutCache(List<Integer/*KType*/> _outFailKeys, Map<Integer/*KType*/, Profile/*VType*/> dtMap, com.vng.zing.zcommon.thrift.PutPolicy policy) {
		return _innerCache.multiPut(_outFailKeys, dtMap, com.vng.zing.common.PutPolicy.fromTPutPolicy(policy));
	}

	/**
	 * multiRemoveCache
	 *
	 * @param keys
	 * @return multi-remove items from cache
	 */
	//////@Override
	public boolean multiRemoveCache(List<Integer/*KType*/> keys) {
		return _innerCache.multiRemove(keys);
	}

	/**
	 * multiRemoveCache
	 *
	 * @param _outFailKeys
	 * @param keys
	 * @return multi-remove items from cache
	 */
	//////@Override
	public boolean multiRemoveCache(List<Integer/*KType*/> _outFailKeys, List<Integer/*KType*/> keys) {
		return _innerCache.multiRemove(_outFailKeys, keys);
	}

	/**
	 * multiPopCache
	 *
	 * @param keys
	 * @return multi-pop items from cache
	 */
	//////@Override
	public Map<Integer/*KType*/, Profile/*VType*/> multiPopCache(List<Integer/*KType*/> keys) {
		return _innerCache.multiPop(keys);
	}

	/**
	 * multiPopCache
	 *
	 * @param _outFailKeys
	 * @param keys
	 * @return multi-pop items from cache
	 */
	//////@Override
	public Map<Integer/*KType*/, Profile/*VType*/> multiPopCache(List<Integer/*KType*/> _outFailKeys, List<Integer/*KType*/> keys) {
		return _innerCache.multiPop(_outFailKeys, keys);
	}

	///~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
	///remote read data methods
	///

	
}
